Day 4 | 限流器 Lab 4 實作：Leaky Bucket＆流量整形

17th鐵人賽

y___che

2025-08-25 18:14:33

138 瀏覽

分享至

前言

前一篇提到了跟時間窗口特色迥異的 Token Bucket 算法，今天要來介紹的是另一個 Bucket 算法，Leaky Bucket 實作起來跟 Token Bucket 其實蠻雷同，桶子的資料結構也很類似，但其特性卻頗有差異，接下來就一起來看看 Leaky Bucket 的設計哲學和工作原理吧。

實作 Lab 4：Leaky Bucket 限流算法

在實作 Leaky Bucket 前，我們用一個真實世界會發生的情況來類比，先建立起高層次的理解。首先，讓我們想像後端服務是一家夜店，圍繞這家夜店的可以分為三群人：

成功進去夜店的人
在夜店外排隊的人
排不進去隊伍就閃的人

然後再想像下有兩根紅龍一前一後佇立在排隊隊伍的頭尾，這個框起來的範圍就是所謂的 Bucket，而紅龍旁邊各站了一個保安，隊伍開頭的那個保安會依照上頭指示，以固定的速率放入固定的人數，這個把人放進來的動作就叫做 Leak。後面那個保安會看看紅龍裡的尾段是不是空出了位置可以把人塞進來，這時有新的人想過來排隊，保安確認有空位就給他排進來，沒有的話就叫他滾蛋。如果你腦中可以想像出這個過程，那基本上 Leaky Bucket 算法已經學完了。

流量整形 Traffic Shaping

上面的這個舉例牽涉到一個重要概念——流量整形（Traffic Shaping），流量整形要解決的問題是，從外部進來的流量頻率可能很不固定，有時會有突如其來的流量高峰，但透過建立一個暫存請求的緩衝區，從這緩衝區以固定的速率對流量放行，讓請求進入到後端服務時表現得很平穩，就像剛才的紅龍區一樣，它形成了一個隊列（Queue）來暫存突發的請求，再以固定的速率消化（Leak）這些請求，至於超出紅龍的人流將會被丟棄，直到隊列的請求消化出空間為止。而這流量整形的概念正是 Leaky Bucket 的設計哲學與核心精神。

Bucket 物件

如果把夜店門外的場景轉換成 Leaky Bucket 的資料結構就會像下面這樣，其實跟 Token Bucket 長得很像，但 Token Bucket 在一開始初始化時就把 Token 裝滿了，因為它的原理中 Bucket 是發 Token 的角色，而 Leaky Bucket 初始化時 Bucket 是空的，等請求進來才會慢慢裝滿，然後照固定速率把裝進來的請求放出去，總而言之參數如下：

capacity：桶子容量
leakRate：請求漏出、抵達服務器的速率
currentRate：當前桶裡的請求數
lastLeakTime：上一次請求漏出的時間

@Getter
@Setter
public class LeakyBucket {

    private final int capacity;
    private final double leakRate;
    private double currentTasks;
    private long lastLeakTime;

    public LeakyBucket(int capacity, double leakRate) {
        this.capacity = capacity;
        this.leakRate = leakRate;
        this.currentTasks = 0;
        this.lastLeakTime = System.currentTimeMillis();
    }
}

計數器

至於計數器的實作，大體上架構也與 Token Bucket 雷同，都是預先定義一個 ConcurrentHashMap<String, LeakyBucket> storage ，並且使用 synchronized 來確保多線程間操作的 Bucket 的原子性，差別僅在於一個是「消費」桶中的東西，一個是把東西「填入」桶中。Leaky Bucket 算法會在把請求裝入桶子前，先檢查應該要放多少請求進去後端服務處理，把應該「漏」的東西漏一漏，漏完才能知道桶裡是否有空間可以裝新的請求。var leakTasks = (elapsedTime / 1000.0) * bucket.getLeakRate(); 這段是計算每秒要漏多少請求的計算，elapsedTime 是從上一次漏水到現在經過多少時間，換算成秒之後乘以漏水的速率得出當下要漏出多少請求。

@Override
public boolean tryAddToLeakyBucket(String key, int capacity, double leakRate) {
    var bucket = leakyBucketStorage.computeIfAbsent(key, k -> new LeakyBucket(capacity, leakRate));

    synchronized (bucket) {
        leakFromBucket(bucket);
        if (bucket.getCurrentTasks() < bucket.getCapacity()) {
            bucket.setCurrentTasks(bucket.getCurrentTasks() + 1);
            return true;
        }
    }
    return false;
}

private void leakFromBucket(LeakyBucket bucket) {
    var now = System.currentTimeMillis();
    var elapsedTime = now - bucket.getLastLeakTime();

    if (elapsedTime > 0) {
        var leakTasks = (elapsedTime / 1000.0) * bucket.getLeakRate();
        var newTasks = Math.max(0, bucket.getCurrentTasks() - leakTasks);
        bucket.setCurrentTasks(newTasks);
        bucket.setLastLeakTime(now);
    }
}

Limiter Strategy 實例

這邊也跟 Token Bucket 的實作方式並無二致，rateLimiter.limit() 是定義桶子容量，rateLimiter.window() 是定義多久漏一次，相除後得出每秒漏多少的速率。

@RequiredArgsConstructor
@Component
public class LeakyBucketLimiter implements RateLimiterStrategy {

    private final RateLimiterStorage rateLimiterStorage;

    @Override
    public boolean isAllow(String key, RateLimiter rateLimiter) {
        var capacity = rateLimiter.limit();
        var leakRate = (double) rateLimiter.limit() / rateLimiter.window();
        return rateLimiterStorage.tryAddToLeakyBucket(key, capacity, leakRate);
    }

    @Override
    public Algorithm getAlgorithmType() {
        return Algorithm.LEAKY_BUCKET;
    }

}

總結

由於 Leaky Bucket 跟 Token Bucket 實作起來十分相似，所以今天的文章放比較多的重點在前面對 Leaky Bucket 特性的描寫，我很喜歡這個算法的設計哲學，它巧妙的運用 Queue 來控制流量的速率，這點是跟其他限流算法非常不同的地方，它可能是對於系統層面最友善的限流算法，因為有它的保護後端基本上不會被擊垮，但相對的就不適用於對用戶體驗要求高的功能，因為它的平穩特性使其失去靈活，就算請求大量湧入也會依照固定的速率響應，增加用戶等待的時間，甚至還會有請求丟失的可能性。